package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/2/14 13:48
 */
public class Dwd_05_DwdTradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_05_DwdTradePayDetailSuc().init(
            3005,
            2,
            "Dwd_05_DwdTradePayDetailSuc"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_05_DwdTradePayDetailSuc");
        readBaseDic(tEnv);
        
        // 2. 从 ods_db 中过滤出支付成的数据
        Table payment_info = tEnv.sqlQuery("select " +
                                               "data['user_id'] user_id, " +
                                               "data['order_id'] order_id, " +
                                               "data['payment_type'] payment_type, " +
                                               "data['callback_time'] callback_time, " +
                                               "et, " +
                                               "pt, " +
                                               "ts " +
                                               "from ods_db " +
                                               "where `database`='gmall2022' " +
                                               "and `table`='payment_info' " +
                                               "and `type`='update' " +
                                               "and `old`['payment_status']='1601' " +
                                               "and `data`['payment_status']='1602' ");
        tEnv.createTemporaryView("payment_info", payment_info);
        
        // 3. 读取下单事务事实表
        tEnv.executeSql("create table dwd_trade_order_detail( " +
                            "id string, " +
                            "order_id string, " +
                            "user_id string, " +
                            "sku_id string, " +
                            "sku_name string, " +
                            "province_id string, " +
                            "activity_id string, " +
                            "activity_rule_id string, " +
                            "coupon_id string, " +
                            "date_id string, " +
                            "create_time string, " +
                            "source_id string, " +
                            "source_type string, " +
                            "source_type_name string, " +
                            "sku_num string, " +
                            "split_original_amount string, " +
                            "split_activity_amount string, " +
                            "split_coupon_amount string, " +
                            "split_total_amount string, " +
                            "ts bigint," +
                            "et as to_timestamp_ltz(ts, 0)," +
                            "watermark for et as et - interval '3' second " +
                            ")" + SQLUtil.getDDLKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_04_DwdTradeCancelDetail"));
        // 4. 3张表的 join
        Table result = tEnv.sqlQuery("select " +
                                         "od.id order_detail_id, " +
                                         "od.order_id, " +
                                         "od.user_id, " +
                                         "od.sku_id, " +
                                         "od.sku_name, " +
                                         "od.province_id, " +
                                         "od.activity_id, " +
                                         "od.activity_rule_id, " +
                                         "od.coupon_id, " +
                                         "pi.payment_type payment_type_code, " +
                                         "dic.dic_name payment_type_name, " +
                                         "pi.callback_time, " +
                                         "od.source_id, " +
                                         "od.source_type source_type_code, " +
                                         "od.source_type_name, " +
                                         "od.sku_num, " +
                                         "od.split_original_amount, " +
                                         "od.split_activity_amount, " +
                                         "od.split_coupon_amount, " +
                                         "od.split_total_amount split_payment_amount, " +
                                         "pi.ts  " +
                                         "from payment_info pi " +
                                         "join dwd_trade_order_detail od " +
                                         "on pi.order_id=od.order_id " +
                                         "and od.et>=pi.et - interval '30' minute and od.et<=pi.et + interval '5' second " +
                                         "join base_dic for system_time as of pi.pt as dic " +
                                         "on pi.payment_type=dic.dic_code ");
        
        // 5. 写出到 kafka
        tEnv.executeSql("create table dwd_trade_pay_detail_suc(  " +
                                "order_detail_id string,  " +
                                "order_id string,  " +
                                "user_id string,  " +
                                "sku_id string,  " +
                                "sku_name string,  " +
                                "province_id string,  " +
                                "activity_id string,  " +
                                "activity_rule_id string,  " +
                                "coupon_id string,  " +
                                "payment_type_code string,  " +
                                "payment_type_name string,  " +
                                "callback_time string,  " +
                                "source_id string,  " +
                                "source_type_code string,  " +
                                "source_type_name string,  " +
                                "sku_num string,  " +
                                "split_original_amount string,  " +
                                "split_activity_amount string,  " +
                                "split_coupon_amount string,  " +
                                "split_payment_amount string,  " +
                                "ts bigint " +
                                ")" + SQLUtil.getDDLKafkaSink(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));
    
        result.executeInsert("dwd_trade_pay_detail_suc");
    }
}
/*
交易域支付成功事务事实表

payment_info 的支付成功数据
    update  : payment_status 1601->1602

    
下单事务事实表 进行 join: interval join

字典表
    退化: payment_type
    

使用 Interval join:
    1. 目前只支持事件时间
        两张表都必须有事件时间
    2. 考虑下上下届
    


 */